package com.zdp.rocketmq.consumer;

import com.zdp.rocketmq.config.JmsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author sesshomaru
 * @date 2021/6/21
 */
@Slf4j
@Component
public class Consumer implements InitializingBean {

    private DefaultMQPushConsumer consumer;

    private String consumerGroup = JmsConfig.GENERAL_CONSUMER_GROUP;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 设置消费者的组名
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setInstanceName("consumer1");
        // 设置地址
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        // 设置如何开始消费消息(消费策略) CONSUME_FROM_LAST_OFFSET 从最后一个开始消费消息
        // CONSUME_FROM_LAST_OFFSET是默认设置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 设置消费模式 默认是集群,消费模式一旦确定就不要修改
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 设置监听什么主题 以这个主题下的哪些标签
        consumer.subscribe(JmsConfig.GENERAL_TOPIC, "*");
        // 这是并发的消费MessageListenerConcurrently
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt msg = list.get(0);
                String keys = null;
                String msgId = null;
                try {
                    log.info("{} Receive New Messages: {}", Thread.currentThread().getName(), new String(msg.getBody()));
                    String topic = msg.getTopic();
                    String body = new String(msg.getBody());
                    String tags = msg.getTags();
                    keys = new String(msg.getBody());
                    msgId = msg.getMsgId();
                    log.info("consumer1 消费消息: topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    // 消费重试次数
                    int reconsumeTimes = msg.getReconsumeTimes();
                    log.error("消费重试次数:{}", reconsumeTimes);
                    log.error("消费异常 keys:{},messageId:{}", keys, msgId);
                    if (reconsumeTimes == 1) {
                        //todo 如果该条消息的重试次数大于1则强行通过这条消息不再让其重试，并记录失败消息日志到数据库
                        log.error("已超过最大消费重试次数 keys:{},messageId:{}", keys, msgId);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        consumer.start();
        log.info("consumer 实例:{} start.....", "conumser1");
    }
}
